package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的所有数据
 * 该类后续提供的一些方法，可能会在多线程的环境下使用，因此要注意线程安全问题，使用 ConcurrentHashMap
 */
public class MemoryDataCenter {
    //交换机：key是 exchangeName，value 是 Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap=new ConcurrentHashMap<>();
    //队列：key是 queueName，value 是 MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap=new ConcurrentHashMap<>();
    //绑定：第一个key是 exchangeName,第二个key是queueName,value是Binding对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap=new ConcurrentHashMap<>();
    //消息：key是messageId，value是Message对象
    private ConcurrentHashMap<String, Message> messageMap=new ConcurrentHashMap<>();
    //队列和消息之间的关系：key是queueName，value是存储了Message对象的LinkedList
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap=new ConcurrentHashMap<>();
    //表示“未被确认”的消息：第一个key是 queueName,第二个key是 MessageId,value是 Message 对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String,Message>> queueMessageWaitAckMap=new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功！exchangeName="+exchange.getName());
    }
    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName="+exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功！queueName="+queue.getName());
    }
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName="+queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        //先根据exchangeName查一下，对应的哈希表是否存在，不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap=bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            bindingMap=new ConcurrentHashMap<>();
            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }
        //上述代码可以写成下面这样
//        ConcurrentHashMap<String, Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),k->new ConcurrentHashMap<>());
        synchronized (bindingMap) {
            //再根据queueName查一下，如果已经存在就抛出异常，不存在插入
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在！ exchangeName=" + binding.getExchangeName() + ",queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功！exchangeName=" + binding.getExchangeName() + ",queueName=" + binding.getQueueName());
    }
    //获取绑定，写两个版本
    //1. 根据 exchangeName 和 queueName 确定唯一一个binding
    //2. 根据 exchangeName 获取到所有的binding
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null){
            return null;
        }
        return bindingMap.get(queueName);
    }
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }
    //删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            //该交换机没有绑定任何队列，报错
            throw new MqException("[MemoryDataCenter]绑定不存在！ exchangeName="+binding.getExchangeName() + ",queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功！exchangeName=" + binding.getExchangeName() + ",queueName=" + binding.getQueueName());

    }

    //添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功！messageId="+message.getMessageId());
    }
    //根据id查询消息
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }
    //根据id删除消息
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除！messageId="+messageId);

    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        //把消息放到对应的队列数据结构中
        //先根据队列的名字，找到该队列对应的消息链表
        LinkedList<Message> messages=queueMessageMap.computeIfAbsent(queue.getName(),k->new LinkedList<>());
        //再把消息添加到messages里面
        synchronized (messages){
            messages.add(message);
            //在这里也把该消息野王消息中心插入一下
        }
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列！ messageId="+message.getMessageId());
    }
    //从队列中取消息
    public Message pollMessage(String queueName){
        //根据队列名查找，对应的队列的消息链表
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        if(messages==null){
            return null;
        }
        synchronized (messages){
            if(messages.size()==0){
                return null;
            }
            //如果有元素，就进行头删
            Message currentMessage=messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中去除！ messageId="+currentMessage.getMessageId());
            return currentMessage;
        }
    }
    //获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        if(messages==null){
            //队列中没有消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    //添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.computeIfAbsent(queueName,k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列！ messageId="+message.getMessageId());
    }
    //删除未确认的消息（消息已经确认了）
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if(messageHashMap==null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除！ messageId="+messageId);
    }
    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if(messageHashMap==null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    //从硬盘上读取数据，把硬盘之前持久化存储的各个维度的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0. 清空之前所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange exchange: exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for(MSGQueue queue:queues){
            queueMap.put(queue.getName(),queue);
        }
        //3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding:bindings){
            ConcurrentHashMap<String, Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //4. 恢复所有的消息数据
        //   遍历所有队列
        for(MSGQueue queue:queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for(Message message:messages) {
                messageMap.put(message.getMessageId(),message);

            }
        }
        // 注意!! 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复. 之前考虑硬盘存储的时候, 也没设定这一块.
        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
        // 这个消息在硬盘上存储的时候, 就是当做 "未被取走"
    }

}
